package com.atguigu.flink.state;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/2/1
 *
 *      广播状态仅仅适用于两个流操作，
 *              一个是数据流，数据流读取配置流中的配置信息。
 *                  一旦配置流的配置信息更新了，需要把更新的配置信息广播到所有的数据流！
 *              一个是配置流(广播流)
 */
public class Demo2_BroadCastState
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //数据流
        SingleOutputStreamOperator<WaterSensor> dataDS = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction());

        //配置流
        SingleOutputStreamOperator<Myconfig> configDS = env
            .socketTextStream("hadoop103", 8889)
            .map(new MapFunction<String, Myconfig>()
            {
                @Override
                public Myconfig map(String value) throws Exception {
                    String[] words = value.split(",");
                    return new Myconfig(words[0], words[1]);
                }
            });

        //普通的流无法使用广播状态，必须是广播流
        MapStateDescriptor<String, String> mapStateDescriptor = new MapStateDescriptor<String, String>("config",String.class,String.class);
        BroadcastStream<Myconfig> broadcastDS = configDS.broadcast(mapStateDescriptor);
        
        //connect 连接两个流
        dataDS.connect(broadcastDS)
              .process(new BroadcastProcessFunction<WaterSensor, Myconfig, WaterSensor>()
              {
                  //处理数据流
                  @Override
                  public void processElement(WaterSensor value, ReadOnlyContext ctx, Collector<WaterSensor> out) throws Exception {

                      //获取最新的配置信息
                      ReadOnlyBroadcastState<String, String> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                      value.setId(broadcastState.get(value.getId()));
                      out.collect(value);

                  }

                  //处理配置流(广播流)
                  @Override
                  public void processBroadcastElement(Myconfig value, Context ctx, Collector<WaterSensor> out) throws Exception {
                        //没收到一条新的配置信息，存入广播状态
                      BroadcastState<String, String> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                      //BroadcastState 当map用
                      broadcastState.put(value.id,value.name);

                  }
              })
              .print();


        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Myconfig{
        private String id;
        private String name;
    }
}
